package com.atguigu.sparkstreaming.utils

import com.atguigu.realtime.utils.PropertiesUtil
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent

/**
 * Created by Smexy on 2022/7/18
 */
object DStreamUtil {

  /*
      如果要把offset提交到kafka:
            createDStream(groupId:String, streamingContext:StreamingContext, topic:String)

      如果把offset提交到mysql:
             createDStream(groupId:String, streamingContext:StreamingContext, topic:String,true,offsetsMap )
   */
  def createDStream(groupId:String, streamingContext:StreamingContext, topic:String,
                    isSaveOffsetToMysql:Boolean = false,offsetsMap: Map[TopicPartition, Long]=null):InputDStream[ConsumerRecord[String, String]]={

    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> PropertiesUtil.getProperty("kafka.broker.list"),
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> groupId,
      //只影响第一次消费。第一次消费当前组没有指定offsets，才读取次参数。如果提交过了偏移量，下一次消费会从已经提交过的位置继续往后消费。
      // 再读取此参数，会更改为none  latest
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> "false"
    )


    val topics = Array(topic)

    var ds: InputDStream[ConsumerRecord[String, String]] = null

    if (isSaveOffsetToMysql){
      ds= KafkaUtils.createDirectStream[String, String](
        streamingContext,
        PreferConsistent,
        Subscribe[String, String](topics, kafkaParams,offsetsMap)
      )

    }else{
      ds= KafkaUtils.createDirectStream[String, String](
        streamingContext,
        PreferConsistent,
        Subscribe[String, String](topics, kafkaParams) )

    }

    ds

  }



}
